package com.weishi.pay.config.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Class_Name: ProducerConfiguration
 * Package: com.weishinfo.center.user.rpc.service.config.mq
 * Describer: MQ消息生产者配置类
 *
 * @author chixiahua
 * @date 2018/7/3 18:27
 **/
@Configuration
@EnableConfigurationProperties(RocketMqProperties.class)
@ConditionalOnProperty(prefix = RocketMqProperties.MQ_CONF_PREFIX, value = "nameServerAddr")
public class ProducerConfiguration {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerConfiguration.class);

    /**
     * MQ配置属性类
     */
    private RocketMqProperties rocketMqProperties;

    public ProducerConfiguration(RocketMqProperties rocketMqProperties) {
        this.rocketMqProperties = rocketMqProperties;
    }

    @Bean
    @ConditionalOnProperty(prefix = RocketMqProperties.MQ_CONF_PREFIX, value = "producerInstanceName")
    public DefaultMQProducer userMQProducer() throws MQClientException {
        DefaultMQProducer userMQProducer = new DefaultMQProducer(rocketMqProperties.getProducerGroupName());
        userMQProducer.setNamesrvAddr(rocketMqProperties.getNameServerAddr());
        userMQProducer.setInstanceName(rocketMqProperties.getProducerInstanceName());

        /**
         * 启动生产者
         */
        userMQProducer.start();
        LOGGER.info("rocketMQ producer 'userMQProducer' started.");
        return userMQProducer;
    }

    @Bean
    @ConditionalOnProperty(prefix = RocketMqProperties.MQ_CONF_PREFIX, value = "consumerInstanceName")
    public DefaultMQPushConsumer userMQConsumer(UserMessageListenerConcurrently userMessageListenerConcurrently) throws MQClientException {
        DefaultMQPushConsumer userMQConsumer = new DefaultMQPushConsumer(rocketMqProperties.getConsumerGroupName());
        userMQConsumer.setNamesrvAddr(rocketMqProperties.getNameServerAddr());
        userMQConsumer.setInstanceName(rocketMqProperties.getConsumerInstanceName());
        if (rocketMqProperties.isConsumerBroadcasting()) {
            // 设置消费模式,默认为集群模式
            userMQConsumer.setMessageModel(MessageModel.BROADCASTING);
        }

        // 设置批量消费数量,默认为1
        userMQConsumer.setConsumeMessageBatchMaxSize(rocketMqProperties.getConsumerBactchMaxSize().equals(0) ? 1 : rocketMqProperties.getConsumerBactchMaxSize());

        // 订阅配置的topic、tag
        for (String s : rocketMqProperties.getSubscribers()) {
            userMQConsumer.subscribe(s.split(":")[0], s.split(":")[1]);
        }

        // 注册消息监听
        userMQConsumer.registerMessageListener(userMessageListenerConcurrently);

        // 启动消费
        userMQConsumer.start();
        LOGGER.info("rocketMQ consumer started.");

        return userMQConsumer;
    }
}
